package com.techsun.handler;

import com.alibaba.fastjson.JSONObject;
import com.mysql.cj.util.StringUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.source.http.HTTPSourceHandler;
import org.apache.log4j.Logger;

import javax.servlet.http.HttpServletRequest;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.*;

public class GeneralHTTPHandler implements HTTPSourceHandler {

    Logger logger = Logger.getLogger(GeneralHTTPHandler.class);

    private static final long serialVersionUID = 4359709211352400082L;

    private static final String charset = "UTF-8";
    private String customer_topic;
    private String retailevent_topic;
    private final JSONObject projectTopicMap = new JSONObject();


    public void configure(Context context) {
        customer_topic = context.getString("customer_topic");
        retailevent_topic = context.getString("retailevent_topic");
        String projectTopicMapStr = context.getString("project_topic_map", "");
        if (!"".equals(projectTopicMapStr)) {
            for (String mapPairStr : projectTopicMapStr.split(";")) {
                String[] split = mapPairStr.split(":");
                projectTopicMap.put(split[0], split[1]);
            }
        }
    }

    public List<Event> getEvents(HttpServletRequest request) throws Exception {
        if (request.getMethod().equalsIgnoreCase("POST")) {
            List<Event> events = new LinkedList<>();
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(request.getInputStream(), charset));
            StringBuilder sb = new StringBuilder();
            String inputStr;
            while ((inputStr = bufferedReader.readLine()) != null) {
                sb.append(inputStr);
            }
            logger.info(String.format("==== POST request ==== \n host:[%s] body:[%s]", request.getRemoteHost(), sb));

            JSONObject jsonObject = JSONObject.parseObject(sb.toString());
            // 获取租户id
            logger.info("projectTopicMap: " + projectTopicMap);
            String mapStr = projectTopicMap.getString(jsonObject.getString("project"));
            String customerTopic;
            String retailEventTopic;
            if (StringUtils.isNullOrEmpty(mapStr)) {
                customerTopic = customer_topic;
                retailEventTopic = retailevent_topic;
            } else {
                customerTopic = customer_topic + mapStr;
                retailEventTopic = retailevent_topic + mapStr;
            }
            if (!jsonObject.containsKey("posted_time")) {
                jsonObject.put("posted_time", System.currentTimeMillis());
            }
            Map<String, String> map = new HashMap<>();
            if ("track_signup".equalsIgnoreCase(jsonObject.getString("event_type"))) {
                map.put("topic", customerTopic);
                events.add(EventBuilder.withBody(jsonObject.toString().getBytes(charset), map));
            }
            map.put("topic",retailEventTopic);
            map.put("key",jsonObject.getString("customer_id"));
            events.add(EventBuilder.withBody(jsonObject.toString().getBytes(charset), map));
            return events;
        }
        return null;
    }
}
